package com.lin.reactor;

import org.junit.jupiter.api.Test;
import reactor.core.Disposable;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;

import java.io.IOException;
import java.time.Duration;
import java.time.temporal.ChronoUnit;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.HashMap;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * @author: linjiang
 * @date: 2025/6/22
 * reactor操作符的功能
 */
public class ReactorOperate {
    /*---------------------------------映射类操作符--------------------------------*/
    @Test
    public void testMap() {
        Flux<Integer> flux = Flux.range(1, 100);
        // 映射
        flux.map(i -> "元素：" + i + " ")
                .subscribe(System.out::print);
    }

    @Test
    public void testFlatMap() {
//        List<Integer> list = Arrays.asList(1, 2, 3, 4, 5, 5);
//        List<String> listStr = Arrays.asList("I", "LOVE", "PROFGRANMMING");

        // 默认的扁平化处理。并发数-16，最大预取数-8
//        Flux.just(list, listStr)
//                .flatMap(var -> Flux.fromIterable(var))
//                .subscribe(System.out::println);

        // 异步线程的时候，默认一次并发10个数据
//        Flux.range(1, 100)
//                .flatMap(var -> Mono.fromCallable(() -> {
//                    Thread.sleep(1000);
//                    return var;
//                }).subscribeOn(Schedulers.boundedElastic()), 10)
//                .log()
//                .blockLast();

        // 异步线程的时候，默认一次并发2个数据，在缓存3个数据。不好测。
//        Flux.range(1, 100)
//                .flatMap(var -> Mono.fromCallable(() -> {
//                    Thread.sleep(1000);
//                    return var;
//                }).subscribeOn(Schedulers.boundedElastic()), 2, 3).log()
//                .blockLast();


        // flatMapDelayError 将计算的错误延时到方法执行完成之后在统一抛出。
        Flux.range(1, 100)
                .flatMapDelayError(var -> Mono.fromCallable(() -> {
                    Thread.sleep(1000);
                    if (var % 2 == 0) {
                        int i = 1 / 0;
                    }
                    return var;
                }).subscribeOn(Schedulers.boundedElastic()), 10, 10).log()
                .doOnError(error -> {
                    System.out.println(error.getMessage());
                })
                .blockLast();

    }

    /*--------------------------------收集类操作符--------------------------------*/
    @Test
    public void testCollect() {
        // 通用型
        Flux<Integer> flux = Flux.range(1, 100);
        flux.collect(Collectors.toList())
                .subscribe(System.out::println);

        // 给定一个容器，将数据收集到该容器中。
        flux.collect(() -> new ArrayList<Integer>(), (list, item) -> list.add(item))
                .subscribe(System.out::println);

        // 直接收集为List
        flux.collectList()
                .subscribe(System.out::println);

        // 直接收集为map，使用给出的key
        flux.collectMap(item -> "item" + item)
                .subscribe(System.out::println);

        // 直接收集为map，给出key 和 value
        flux.collectMap(item -> "item" + item, item -> item * 2)
                .subscribe(System.out::println);

        // 给出key 和 value，收集为给定的容器。
        flux.collectMap(item -> "item" + item, item -> item * 3, () -> new ConcurrentHashMap<>(16))
                .subscribe(i -> {
                    if (i instanceof ConcurrentHashMap) {
                        System.out.println(true);
                    }
                    System.out.println(i);
                });

        // 将元素按奇偶性进行收集
        flux.collectMultimap(item -> item % 2 == 0 ? "偶数" : "奇数")
                .subscribe(System.out::println);

        // 将元素倒序排列后收集
        flux.collectSortedList(Comparator.reverseOrder())
                .subscribe(System.out::println);

    }

    /*----------------------------统计函数-------------------------------*/
    @Test
    public void testStatistic() {
        Flux<Integer> flux = Flux.range(1, 100);

        // 求和
        flux.reduce(0, (a, b) -> a + b)
                .subscribe(System.out::println);

        // 统计数量
        flux.count().subscribe(System.out::println);
    }


    /*--------------------------分组函数-----------------------------*/
    @Test
    public void testGroupBy() {
        Flux<Integer> flux = Flux.range(1, 100);

        // 自定义key，进行分组.
        flux.groupBy(item -> item % 2 == 0 ? "偶数" : "奇数")
                .subscribe(i -> {
                    // 对分组之后的数据进行收集。这个就相当于java9中Processer。
                    i.collectList()
                            .subscribe(System.out::println);
                });

        // 自定义key，变更value值
        flux.groupBy(key -> key % 2 == 0 ? "偶数" : "奇数", value -> value * 2)
                .subscribe(i -> {
                    i.collectList()
                            .subscribe(System.out::println);
                });

        flux.groupBy(i -> i % 2 == 0 ? "even" : "odd")
                .flatMap(g -> g.collectList()
                        .map(list -> new HashMap() {{
                            put(g.key(), list);
                        }}))
                .subscribe(i -> {
                    System.out.println(i);
                });
    }


    /*---------------------------过滤函数--------------------------------*/
    @Test
    public void testFilter() {
        Flux<Integer> flux = Flux.range(1, 10);

        // 同步过滤
        flux.filter(item -> item % 2 == 0)
                .log()
                .subscribe();

        // 异步过滤
        flux.filterWhen(item -> Mono.just(item % 2 == 0))
                .log()
                .subscribe();

    }


    /*-----------------------------高阶函数-----------------------------------*/

    @Test
    public void testTransform() {
        Function<Flux<String>, Flux<Object>> transformFun =
                f -> f.filter(item -> !item.equals("red")).map(item -> item.toUpperCase());

        Flux<String> flux = Flux.just("red", "blue", "yellow");

        // transform 函数，允许将操作链中的部分封装成一个函数。从而使相同的序列复用封装的函数。
        flux.doOnNext(System.out::println)
                .transform(transformFun)
                .subscribe(i -> System.out.println("最终的元素:" + i));
    }

    @Test
    public void testClodAndHot() throws InterruptedException, IOException {
        Flux<Long> flux = Flux.interval(Duration.of(1, ChronoUnit.SECONDS))
                /**
                 * 冷流：
                 *  不同时间订阅的订阅者，都会接收到来自发布者发布的全部数据。
                 * 热流:
                 *  不同时间订阅的订阅者，只会收到来自发布者开始发布的数据。
                 *  share() -> 将冷流转换未热流
                 */
                .share();

        // 订阅者1
        flux.subscribe(i -> System.out.println("订阅者1 ->" + i));

        TimeUnit.SECONDS.sleep(10);
        flux.subscribe(i -> System.out.println("订阅者2 ->" + i));

        System.in.read();
    }

    @Test
    public void testConnectableFlux() throws InterruptedException, IOException {
        /**
         * ConnectableFlux:能控制流什么时候发布
         */
        Flux<Long> flux = Flux.interval(Duration.of(1, ChronoUnit.SECONDS));

        // 1.connect API 手动连接，并发送发布数据
//        ConnectableFlux<Long> publish = flux.publish();
//        publish.subscribe(i -> System.out.println("订阅者1 ->" + i));
//        publish.subscribe(i -> System.out.println("订阅者2 ->" + i));
//        // 在connect之前的数据，都会接收到flux发布的数据，但是在connect之后，才会收到后续发布的数据。
//        publish.connect();
//        TimeUnit.SECONDS.sleep(5);
//        publish.subscribe(i -> System.out.println("订阅者3 ->" + i));

        // 2.autoConnect(N) API 当订阅者达到N时，然后自动发送发布数据
        Flux<Long> publish = flux.publish().autoConnect(2);
        Disposable d1 = publish.subscribe(i -> System.out.println("订阅者1 ->" + i));
        Disposable d2 = publish.subscribe(i -> System.out.println("订阅者2 ->" + i));
        TimeUnit.SECONDS.sleep(2);
        d1.dispose();
        d2.dispose();
        TimeUnit.SECONDS.sleep(2);
        // 自动发布数据之后，之后订阅的数据会收到后续发布数据
        publish.subscribe(i -> System.out.println("订阅者3 ->" + i));


        // 3.refCount(N) API 至少需要N个订阅者，才会自动发送发布数据。
//        Flux<Long> publish = flux.publish().refCount(2);
//        Disposable d1 = publish.subscribe(i -> System.out.println("订阅者1 ->" + i));
//        Disposable d2 = publish.subscribe(i -> System.out.println("订阅者2 ->" + i));
//        Disposable d3 = publish.subscribe(i -> System.out.println("订阅者3 ->" + i));
//
//        TimeUnit.SECONDS.sleep(2);
//        d1.dispose();
//        d2.dispose();
//        d3.dispose();
//        TimeUnit.SECONDS.sleep(2);
//        publish.subscribe(i -> System.out.println("订阅者4 ->" + i));
//        publish.subscribe(i -> System.out.println("订阅者5 ->" + i));

        System.in.read();
    }

    @Test
    public void testParallel() {
        Flux<Integer> flux = Flux.range(1, 10);

        // 串行
        flux.log().subscribe();
        System.out.println("----------------");
        // 并行
        flux
                // 创建两个并行通道
                .parallel(2)
                // 创建两个线程
                .runOn(Schedulers.parallel())
                .log()
                .subscribe();

    }


}
